package com.caltco.cargo.ops.config;


import com.caltco.cargo.ops.common.utils.RabbitMQEnum;
import com.caltco.cargo.ops.modules.sys.service.impl.EmailServiceImp;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;


/**
 * @Author Bernie
 * @Date 2020/7/15 11:02
 * RabbitMQ 配置文件
 */
@Slf4j
@Configuration
@ConditionalOnProperty(prefix = "spring.rabbitmq", name = "enabled", havingValue = "true")
public class RabbitMQAutoConfig {
    private static final Logger log = LoggerFactory.getLogger(EmailServiceImp.class);

    @Autowired
    private CachingConnectionFactory connectionFactory;
    @Autowired
    private AmqpAdmin amqpAdmin;



    @Bean
    public RabbitTemplate rabbitTemplate(){
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMessageConverter(converter());

        //消息是否成功发送到Exchange
        rabbitTemplate.setConfirmCallback(((correlationData, ack, cause) -> {
            String msgId = correlationData.getId();
            if(ack){
                log.info("correlationData:{}",correlationData.toString());
                log.info("消息成功发送到 Exchange");

            }else{
                log.info("消息发送到Exchange失败，{},cause:{}",correlationData,cause);
            }
        }));
        // 触发setReturnCallback回调必须设置mandatory=true,否则Exchange没有找到Queue就会丢弃消息，
        // 而不会触发回调
        rabbitTemplate.setMandatory(true);
        //消息是否从Exchange路由到Queue,注：这是个失败回调，只有消息从Exchange路由到Queue失败，才会
        // 回調这个方法
        rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
            @Override
            public void returnedMessage(ReturnedMessage returnedMessage) {
                log.info("消息从Exchange路由到Queue失败，exchagne:{},routingKey:{},replyCode:{},replyText:{},message:{}"
                        ,returnedMessage.getExchange(),returnedMessage.getRoutingKey()
                        ,returnedMessage.getReplyCode(),returnedMessage.getReplyText()
                        ,returnedMessage.getMessage());

            }
        });
        return rabbitTemplate;
    }

    @Bean
    public Jackson2JsonMessageConverter converter(){
        return new Jackson2JsonMessageConverter();
    }

    /**
     * 创建用户信息交换器
     * @return
     */
    @Bean
    public TopicExchange alarmTopicExchange(){
        TopicExchange dataExchange = new TopicExchange(RabbitMQEnum.USER_TOPIC_EXCHANGE.getValue(),true,false);
        log.info("创建交换器 {} [{}]", RabbitMQEnum.USER_TOPIC_EXCHANGE.getValue(), dataExchange);
        try {
            amqpAdmin.declareExchange(dataExchange);
        }catch (Exception e){
            System.out.println("连接失败");
        }
        return dataExchange;
    }

    @Bean
    public Queue firstQueue() {
        return new Queue("USER_QUEUE");
    }

    @Bean
    Binding bindingExchangeMessage2() {
        return BindingBuilder.bind(firstQueue()).to(alarmTopicExchange()).with("user.#");
    }


}
